Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

examples/web_scraper_cache/scraper.py (5)

28-29: Avoid configuring root logging at import time.

Calling basicConfig at module level affects host applications and test suites. Move the configuration inside main() or behind a guard like if __name__ == "__main__":.

---

75-85: Enforce size limit after encryption to cap actual stored bytes.

The current order checks size before encryption. Fernet adds overhead, so entries near 5MB may exceed the cap on disk. Wrap LimitSizeWrapper inside FernetEncryptionWrapper so the size check applies to encrypted bytes.

---

95-107: Normalize URLs before hashing to reduce duplicate keys.

Hashing raw URLs treats trivial differences (case, default ports, fragments, trailing slash) as different keys. Consider normalizing URLs (lowercase scheme/host, remove default ports, handle trailing slashes) before hashing to improve cache efficiency.

---

134-145: Harden reads against decryption/validation errors.

A bad key or corrupt entry will raise and bubble to callers. Wrap await self.adapter.get(...) in a try/except to catch exceptions, log the error, and return None instead of propagating the exception.

---

160-171: Consider a cheap existence check.

is_cached() fetches and decodes full values. If the underlying store or adapter exposes an exists()/has() method, prefer that to avoid decrypting/decoding payloads when only presence is needed.

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

examples/chat_app/chat_app.py (1)

134-134: Consider moving logging configuration to the __main__ guard.

While having logging.basicConfig inside main() is better than import-time configuration, the previous review feedback suggested placing it directly in the __main__ guard. This ensures logging setup happens only when running the script directly, not when calling main() programmatically.

 async def main():
     """Demonstrate the chat application."""
-    # Configure logging for the demo
-    logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-
     chat = ChatApp()
     # ...

 if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
     asyncio.run(main())

examples/web_scraper_cache/scraper.py (4)

68-83: Enforce size limit after encryption to cap actual stored bytes.

The current wrapper order checks size before encryption. Since Fernet adds overhead (~57 bytes + the data), entries near 5MB may exceed the intended storage cap. Wrap LimitSizeWrapper inside FernetEncryptionWrapper to enforce the limit on encrypted bytes.

Apply this diff:

-        primary_with_wrappers = TTLClampWrapper(
-            key_value=LimitSizeWrapper(
-                key_value=FernetEncryptionWrapper(
-                    key_value=FallbackWrapper(key_value=disk_store, fallback=fallback_store),
-                    key=encryption_key,
-                ),
-                max_size_bytes=5 * 1024 * 1024,  # 5MB limit
-            ),
-            min_ttl=3600,  # 1 hour minimum
-            max_ttl=7 * 24 * 3600,  # 7 days maximum
-        )
+        primary_with_wrappers = TTLClampWrapper(
+            key_value=FernetEncryptionWrapper(
+                key_value=LimitSizeWrapper(
+                    key_value=FallbackWrapper(key_value=disk_store, fallback=fallback_store),
+                    max_size_bytes=5 * 1024 * 1024,  # 5MB limit (enforced on encrypted bytes)
+                ),
+                key=encryption_key,
+            ),
+            min_ttl=3600,                 # 1 hour minimum
+            max_ttl=7 * 24 * 3600,        # 7 days maximum
+        )

---

93-105: Normalize URLs before hashing to reduce duplicate keys.

Hashing the raw URL treats trivial differences (scheme/hostname case, default ports, fragments, trailing slashes) as distinct keys, causing redundant cache entries.

Apply this diff to add normalization:

+from urllib.parse import urlsplit, urlunsplit
+
 @@
     def _url_to_key(self, url: str) -> str:
         """
         Convert URL to a safe cache key.
 
-        Uses SHA-256 hash to create a consistent, filesystem-safe key.
+        Normalizes the URL then uses SHA-256 hash to create a consistent, filesystem-safe key.
 
         Args:
             url: The URL to convert
 
         Returns:
             Hashed key string
         """
-        return hashlib.sha256(url.encode()).hexdigest()
+        normalized = self._normalize_url(url)
+        return hashlib.sha256(normalized.encode()).hexdigest()
+
+    def _normalize_url(self, url: str) -> str:
+        """Normalize URL to reduce duplicate cache entries."""
+        parts = urlsplit(url)
+        scheme = parts.scheme.lower()
+        netloc = parts.netloc.lower()
+        # Remove default ports
+        if scheme == "http" and netloc.endswith(":80"):
+            netloc = netloc.rsplit(":", 1)[0]
+        elif scheme == "https" and netloc.endswith(":443"):
+            netloc = netloc.rsplit(":", 1)[0]
+        # Ensure non-empty path; drop fragment
+        path = parts.path or "/"
+        return urlunsplit((scheme, netloc, path, parts.query, ""))

---

132-143: Harden reads against decryption/validation errors.

A wrong encryption key or corrupted entry will raise an exception that bubbles to callers. Catch these errors and return None instead.

Apply this diff:

     async def get_cached_page(self, url: str) -> ScrapedPage | None:
         """
         Retrieve a cached page.
 
         Args:
             url: Page URL
 
         Returns:
             ScrapedPage if found and valid, None otherwise
         """
         key = self._url_to_key(url)
-        return await self.adapter.get(collection="pages", key=key)
+        try:
+            return await self.adapter.get(collection="pages", key=key)
+        except Exception as e:
+            logger.error("Failed to read cached page %s: %s", url, e)
+            return None

---

158-169: Consider a cheap existence check.

is_cached() fetches the full value, requiring decryption and deserialization. If the adapter or underlying store exposes an exists() or has() method, use that for O(1) presence checks without payload processing.

If PydanticAdapter or the underlying key-value store provides an existence check method, refactor to:

async def is_cached(self, url: str) -> bool:
    """Check if a page is cached without fetching the full payload."""
    key = self._url_to_key(url)
    # If adapter/store has exists() or has():
    return await self.adapter.exists(collection="pages", key=key)
    # Otherwise fallback to current implementation

examples/trading_data/README.md (2)

45-51: Add working directory context and clarify uv usage.

Run commands from the example folder and note uv is an editable install.

Apply:

 ```bash
-# Install dependencies
-pip install -r requirements.txt
+# From the example directory
+cd examples/trading_data
+
+# Install dependencies
+pip install -r requirements.txt
@@
-# Or using uv
-uv pip install -e .
+# Or using uv (editable install of the example)
+uv pip install -e .

---

`139-145`: **Undefined variable in Retry example; use the defined disk cache.**


Apply:

```diff
 retry_store = RetryWrapper(
-    key_value=base_store,
+    key_value=disk_cache,
     max_retries=3,
     base_delay=0.1
 )

examples/trading_data/trading_app.py (2)

141-141: Return type includes a float (hit_rate_percent).

Apply:

-def get_cache_statistics(self) -> dict[str, int]:
+def get_cache_statistics(self) -> dict[str, int | float]:

---

115-116: Docstring should reflect “latest” alias now being maintained.

Apply:

-        Note: This is a simplified implementation. In production, you'd want
-        to maintain a separate "latest" key or use a time-series database.
+        Note: For demo convenience, store_price also maintains a "latest" alias
+        per symbol. For high write rates, consider a time-series database.

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: CR
PR: strawgate/py-key-value#0
File: AGENTS.md:0-0
Timestamp: 2025-10-26T16:57:34.087Z
Learning: Applies to **/*.py : Use Ruff for Python formatting and linting (configured in pyproject.toml)

examples/chat_app/chat_app.py (2)

74-75: Potential message ID collision with timestamp-based keys.

Using timestamp.isoformat() as the message ID could lead to collisions if multiple messages are sent within the same microsecond (e.g., in automated tests or high-throughput scenarios). While this may be acceptable for a simple example, consider documenting this limitation or appending a UUID suffix for uniqueness.

Example fix:

+import uuid
+
 # ...
 
-        # Use timestamp as message ID for chronological ordering
-        message_id = message.timestamp.isoformat()
+        # Use timestamp + UUID for chronological ordering with uniqueness guarantee
+        message_id = f"{message.timestamp.isoformat()}-{uuid.uuid4().hex[:8]}"

---

120-135: Redundant isinstance check; aggregation logic is sound.

The isinstance check is redundant since self.stats_wrapper is always assigned a StatisticsWrapper instance in __init__ (line 58). However, the statistics aggregation across collections is a good implementation for multi-conversation tracking.

     def get_statistics(self) -> dict[str, int]:
         """
         Get operation statistics across all conversations.
 
         Returns:
             Dictionary with aggregated operation counts (puts, gets, deletes, etc.)
         """
-        if isinstance(self.stats_wrapper, StatisticsWrapper):
-            # Aggregate statistics across all collections (conversations)
-            total_puts = sum(coll_stats.put.count for coll_stats in self.stats_wrapper.statistics.collections.values())
-            total_gets = sum(coll_stats.get.count for coll_stats in self.stats_wrapper.statistics.collections.values())
-            total_deletes = sum(coll_stats.delete.count for coll_stats in self.stats_wrapper.statistics.collections.values())
-            get_hits = sum(coll_stats.get.hit for coll_stats in self.stats_wrapper.statistics.collections.values())
-            get_misses = sum(coll_stats.get.miss for coll_stats in self.stats_wrapper.statistics.collections.values())
-
-            return {
-                "total_puts": total_puts,
-                "total_gets": total_gets,
-                "total_deletes": total_deletes,
-                "get_hits": get_hits,
-                "get_misses": get_misses,
-            }
-        return {}
+        # Aggregate statistics across all collections (conversations)
+        total_puts = sum(coll_stats.put.count for coll_stats in self.stats_wrapper.statistics.collections.values())
+        total_gets = sum(coll_stats.get.count for coll_stats in self.stats_wrapper.statistics.collections.values())
+        total_deletes = sum(coll_stats.delete.count for coll_stats in self.stats_wrapper.statistics.collections.values())
+        get_hits = sum(coll_stats.get.hit for coll_stats in self.stats_wrapper.statistics.collections.values())
+        get_misses = sum(coll_stats.get.miss for coll_stats in self.stats_wrapper.statistics.collections.values())
+
+        return {
+            "total_puts": total_puts,
+            "total_gets": total_gets,
+            "total_deletes": total_deletes,
+            "get_hits": get_hits,
+            "get_misses": get_misses,
+        }

examples/web_scraper_cache/scraper.py (4)

76-87: Enforce size limit after encryption to cap actual stored bytes.

The current wrapper order checks size before encryption. Since Fernet adds overhead (~57 bytes + padding), entries near 5MB plaintext will exceed the limit after encryption.

Apply this diff to move the size check inside the encryption wrapper:

 primary_with_wrappers = TTLClampWrapper(
-    key_value=LimitSizeWrapper(
-        key_value=FernetEncryptionWrapper(
-            key_value=FallbackWrapper(primary_key_value=disk_store, fallback_key_value=fallback_store),
+    key_value=FernetEncryptionWrapper(
+        key_value=LimitSizeWrapper(
+            key_value=FallbackWrapper(primary_key_value=disk_store, fallback_key_value=fallback_store),
+            max_size=5 * 1024 * 1024,  # 5MB limit (applies to encrypted bytes)
+        ),
-            fernet=fernet,
-            raise_on_decryption_error=False,  # Return None on decryption failure instead of raising
-        ),
-        max_size=5 * 1024 * 1024,  # 5MB limit
+        fernet=fernet,
+        raise_on_decryption_error=False,  # Return None on decryption failure instead of raising
     ),
     min_ttl=3600,  # 1 hour minimum
     max_ttl=7 * 24 * 3600,  # 7 days maximum
 )

---

97-109: Consider normalizing URLs before hashing to reduce duplicate keys.

Hashing raw URLs treats trivial differences (scheme case, default ports, fragments, trailing slashes) as different keys, potentially causing unnecessary cache misses.

Apply this diff to add URL normalization:

+from urllib.parse import urlsplit, urlunsplit
+
 ...
 
     def _url_to_key(self, url: str) -> str:
         """
         Convert URL to a safe cache key.
 
-        Uses SHA-256 hash to create a consistent, filesystem-safe key.
+        URLs are normalized before hashing to treat trivial variations
+        (case, default ports, fragments) as identical.
 
         Args:
             url: The URL to convert
 
         Returns:
             Hashed key string
         """
-        return hashlib.sha256(url.encode()).hexdigest()
+        normalized = self._normalize_url(url)
+        return hashlib.sha256(normalized.encode()).hexdigest()
+
+    def _normalize_url(self, url: str) -> str:
+        """Normalize URL to reduce duplicate cache entries."""
+        parts = urlsplit(url)
+        scheme = parts.scheme.lower()
+        netloc = parts.netloc.lower()
+        
+        # Remove default ports
+        if scheme == "http" and netloc.endswith(":80"):
+            netloc = netloc.rsplit(":", 1)[0]
+        elif scheme == "https" and netloc.endswith(":443"):
+            netloc = netloc.rsplit(":", 1)[0]
+        
+        # Normalize path and drop fragment
+        path = parts.path or "/"
+        return urlunsplit((scheme, netloc, path, parts.query, ""))

---

162-173: Consider using a lightweight existence check instead of fetching full values.

is_cached() currently calls get_cached_page(), which fetches, decrypts, and deserializes the entire payload just to check presence. For large pages, this is inefficient.

If the underlying KeyValue interface exposes an exists() or has() method, use it here for O(1) presence checks. Otherwise, you might want to add such a method to PydanticAdapter:

async def is_cached(self, url: str) -> bool:
    """
    Check if a page is cached.
    
    Args:
        url: Page URL
    
    Returns:
        True if page is cached, False otherwise
    """
    key = self._url_to_key(url)
    # If adapter/store supports exists():
    return await self.adapter.exists(collection="pages", key=key)
    # Otherwise, keep current implementation as fallback

Check if the KeyValue interface or PydanticAdapter provides an existence check method:

#!/bin/bash
# Search for exists/has methods in the key-value library

# Look for exists or has method definitions
ast-grep --pattern $'class $CLASS {
  $$$
  async def exists($$$) {
    $$$
  }
  $$$
}'

ast-grep --pattern $'class $CLASS {
  $$$
  async def has($$$) {
    $$$
  }
  $$$
}'

# Also check with ripgrep
rg -nP --type=py 'async def (exists|has)\s*\(' -A 3

---

136-147: Consider defensive error handling for deserialization failures.

While decryption errors return None (line 81), Pydantic validation errors could still raise. Adding a try/except would make the method more resilient.

Apply this diff:

     async def get_cached_page(self, url: str) -> ScrapedPage | None:
         """
         Retrieve a cached page.
 
         Args:
             url: Page URL
 
         Returns:
             ScrapedPage if found and valid, None otherwise
         """
         key = self._url_to_key(url)
-        return await self.adapter.get(collection="pages", key=key)
+        try:
+            return await self.adapter.get(collection="pages", key=key)
+        except Exception:
+            logger.exception(f"Failed to retrieve cached page {url}")
+            return None

examples/trading_data/requirements.txt (1)

1-2: Remove duplication; use pyproject.toml as single source of truth.

These dependencies duplicate lines 7-8 in pyproject.toml. Having both files creates a maintenance burden and risks version drift. Since pyproject.toml is the modern standard (PEP 621), remove this file and use pip install -e . or pip install -e ".[dev]" instead.

examples/trading_data/trading_app.py (2)

78-98: Fix filesystem portability and write "latest" alias.

Two issues need addressing:

1. Windows filesystem incompatibility: Line 94 uses isoformat() which includes colons (e.g., 2024-10-26T12:34:56+00:00), and line 96 uses f"symbol:{symbol}" with a colon. Both are invalid for Windows file paths when using DiskStore.

2. Missing "latest" alias: get_latest_price() expects a "latest" key (line 128) that is never written, making that method non-functional.

Apply this diff:

+    @staticmethod
+    def _collection_for(symbol: str) -> str:
+        """Return filesystem-safe collection name."""
+        return f"symbol-{symbol.replace('/', '_').replace(':', '_')}"
+
     async def store_price(self, symbol: str, price: float, volume: int, ttl: int | None = None) -> str:
         """
         Store price data for a symbol.
 
         Args:
             symbol: Trading symbol (e.g., "AAPL", "BTC-USD")
             price: Current price
             volume: Trading volume
             ttl: Time-to-live in seconds (optional)
 
         Returns:
             Price data ID (timestamp-based)
         """
         price_data = PriceData(symbol=symbol, price=price, volume=volume, timestamp=datetime.now(tz=timezone.utc))
 
-        # Use timestamp as key for chronological ordering
-        data_id = price_data.timestamp.isoformat()
+        # Filesystem-safe, lexicographically sortable key
+        data_id = price_data.timestamp.strftime("%Y%m%dT%H%M%S%fZ")
 
-        await self.adapter.put(collection=f"symbol:{symbol}", key=data_id, value=price_data, ttl=ttl)
+        col = self._collection_for(symbol)
+        await self.adapter.put(collection=col, key=data_id, value=price_data, ttl=ttl)
+        # Maintain "latest" pointer for get_latest_price()
+        await self.adapter.put(collection=col, key="latest", value=price_data, ttl=ttl)
 
         return data_id

---

100-141: Apply filesystem-safe collection naming to all methods.

These methods use f"symbol:{symbol}" which is incompatible with Windows file paths. Use the _collection_for() helper method suggested in the previous comment.

Apply this diff:

     async def get_price(self, symbol: str, data_id: str) -> PriceData | None:
         """
         Retrieve price data for a symbol.
 
         Args:
             symbol: Trading symbol
             data_id: Price data identifier (timestamp)
 
         Returns:
             PriceData if found, None otherwise
         """
-        return await self.adapter.get(collection=f"symbol:{symbol}", key=data_id)
+        return await self.adapter.get(collection=self._collection_for(symbol), key=data_id)
 
     async def get_latest_price(self, symbol: str) -> PriceData | None:
         """
         Get the most recent price data for a symbol.
 
-        Note: This is a simplified implementation. In production, you'd want
-        to maintain a separate "latest" key or use a time-series database.
+        Note: store_price maintains a "latest" alias per symbol for fast lookups.
 
         Args:
             symbol: Trading symbol
 
         Returns:
             Latest PriceData if available, None otherwise
         """
-        # This is a demonstration - in production you'd track the latest key
-        # For now, this just returns None to show the API pattern
-        return await self.adapter.get(collection=f"symbol:{symbol}", key="latest")
+        return await self.adapter.get(collection=self._collection_for(symbol), key="latest")
 
     async def delete_price(self, symbol: str, data_id: str) -> bool:
         """
         Delete price data.
 
         Args:
             symbol: Trading symbol
             data_id: Price data identifier
 
         Returns:
             True if deleted, False if not found
         """
-        return await self.adapter.delete(collection=f"symbol:{symbol}", key=data_id)
+        return await self.adapter.delete(collection=self._collection_for(symbol), key=data_id)

examples/trading_data/pyproject.toml (1)

17-18: Pytest asyncio config added; consider explicit testpaths for consistency.

The asyncio_mode = "auto" has been properly added. For completeness, you could add testpaths = ["."] to make test discovery explicit, but this is optional since pytest will auto-discover tests in the current directory.

Optional addition:

 [tool.pytest.ini_options]
 asyncio_mode = "auto"
+testpaths = ["."]

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro